package org.example.datastream.test.custom;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.example.util.Utils;

import java.math.BigDecimal;
import java.sql.*;


public class EnterpriseJdbcSource extends RichSourceFunction<Enterprise> {

    private PreparedStatement ps;
    private Connection conn;
    private ResultSet rs;

    private String driver;
    private String url;
    private String username;
    private String password;
    private String sql;

    public EnterpriseJdbcSource(String driver, String url, String username, String password, String sql) {
        this.driver = driver;
        this.url = url;
        this.username = username;
        this.password = password;
        this.sql = sql;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        Class.forName(driver);

        conn = DriverManager.getConnection(url, username, password);
        ps = conn.prepareStatement(sql);
    }

    @Override
    public void run(SourceContext<Enterprise> sourceContext) throws Exception {
        rs = ps.executeQuery();
        while(rs.next()) {
            Enterprise item = new Enterprise();
            item.setEid(rs.getString("eid"));
            item.setEnterpriseName(rs.getString("name"));
            item.setStandardName(rs.getString("format_name"));
            item.setGroupName(rs.getString("group_name"));
            item.setGroupAbbreviation(rs.getString("group_abbreviation"));
            item.setLogoUrl(rs.getString("logo_url"));
            item.setOperName(rs.getString("oper_name"));
            item.setOperType(rs.getString("oper_type"));
            item.setOperNameId(rs.getString("oper_name_id"));
            item.setTitle(rs.getString("title"));

            item.setTitleCode(1);//rs.getInt("title_code") //todo:转换逻辑
            item.setRecCapi(rs.getBigDecimal("regist_capi_new"));
            String actualCapi = rs.getString("actual_capi");
            BigDecimal actualCapiDecimal = Utils.getDecimalFromStr(actualCapi);
            item.setRecCapi(actualCapiDecimal);
            item.setCurrencyUnit(rs.getString("currency_unit"));
            item.setBusinessUrl(rs.getString("url"));
            item.setOperName(rs.getString("collegues_num"));
            item.setBelongOrg(rs.getString("belong_org"));
            item.setDepartment(rs.getString("department"));
            item.setScope(rs.getString("scope"));
            item.setRegNo(rs.getString("reg_no"));

            item.setCreditNo(rs.getString("credit_no"));
            item.setOrgNo(rs.getString("org_no"));
            item.setEconKindCode(1);//rs.getInt("econ_kind_code") //todo:转换逻辑
            item.setType(rs.getInt("type"));
            item.setTypeNew(rs.getString("type_new"));
            item.setCategoryNew(rs.getString("category_new"));
            item.setStatus(rs.getString("status"));
            item.setStatusCode(1);//rs.getInt("new_status_code") //todo:转换逻辑
            Date startDate = Utils.str2Date2(rs.getString("start_date"));
            item.setStartDate(startDate);
            Date termStart = Utils.str2Date2(rs.getString("term_start"));
            item.setTermStart(termStart);

            Date termEnd = Utils.str2Date2(rs.getString("term_end"));
            item.setTermEnd(termEnd);
            Date checkDate = Utils.str2Date2(rs.getString("check_date"));
            item.setCheckDate(checkDate);
            Date logoutDate = Utils.str2Date2(rs.getString("logout_date"));
            item.setLogoutDate(logoutDate);
            item.setLogoutReason(rs.getString("logout_reason"));
            Date revokeDate = Utils.str2Date2(rs.getString("revoke_date"));
            item.setRevokeDate(revokeDate);
            item.setRevokeReason(rs.getString("revoke_reason"));
            item.setRevokedCertificates(rs.getString("revoked_certificates"));
            item.setProvinceCode(rs.getString("province_code"));
            item.setSource(rs.getString("source"));
            item.setRowUpdateTime(rs.getTimestamp("row_update_time"));

//            long createdTimeNum = rs.getLong("created_time");
//            LocalDateTime dateTime = Utils.timestamp2Datetime(createdTimeNum);
//            Timestamp timestamp = Timestamp.valueOf(dateTime);
            item.setCreateTime(rs.getTimestamp("row_update_time"));

            sourceContext.collect(item);
        }
    }

    @Override
    public void cancel() {

    }

    @Override
    public void close() throws Exception {
        super.close();

        if (null != conn) {
            conn.close();
        }
        if (null != ps) {
            ps.close();
        }
    }
}
